package com.wudl.flink.doris.source;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.wudl.flink.doris.bean.DorisBean;
import com.wudl.flink.doris.utils.MyKafkaUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author ：wudl
 * @date ：Created in 2022-01-01 13:05
 * @description： 产生数据
 * @modified By：
 * @version: 1.0
 */

public class GenerateData implements SourceFunction<String>  {


        private boolean isRunning = true;

        String[] citys = {"北京","广东","山东","江苏","河南","上海","河北","浙江","香港","山西","陕西","湖南","重庆","福建","天津","云南","四川","广西","安徽","海南","江西","湖北","山西","辽宁","内蒙古"};

        Integer i = 0;
        List<DorisBean>  list = new ArrayList<>();
        @Override
        public void run(SourceContext<String> ctx) throws Exception {

            Random random = new Random();
            SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            while (isRunning) {
                int number = random.nextInt(4) + 1;
                Integer id =  i++;
                String name = df.format(new Date());
//                Integer name = id+100;
                String address = "1";
                String city = citys[random.nextInt(citys.length)];
//                int age = random.nextInt(25);
                String phone = getTel();
                DorisBean  dorisBean = new DorisBean(id,name,address,city,phone);
                list.add(dorisBean);
                if (list.size()==2000)
                {
                    Map<String,List> map = new HashMap<>();
                    map.put("data",list);
                    String s = JSON.toJSONString(map);
                    System.out.println("map--->"+s);
                    list = new ArrayList<>();
                    ctx.collect(s);
//                    Thread.sleep(5000*2);
                }


            }
        }

        @Override
        public void cancel() {

            isRunning = false;
        }


    private static String[] telFirst="134,135,136,137,138,139,150,151,152,157,158,159,130,131,132,155,156,133,153".split(",");
    private static String getTel() {
        int index=getNum(0,telFirst.length-1);
        String first=telFirst[index];
        String second=String.valueOf(getNum(1,888)+10000).substring(1);
        String third=String.valueOf(getNum(1,9100)+10000).substring(1);
        return first+second+third;
    }

    public static int getNum(int start,int end) {
        return (int)(Math.random()*(end-start+1)+start);
    }

    public static void main(String[] args) throws Exception {
         String default_topic = "wudltopicdoris01";
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(8);
        DataStreamSource<String> generateSource = env.addSource(new GenerateData());
        generateSource.print("--------");
        generateSource.addSink(MyKafkaUtil.getKafkaProducer(default_topic));

        env.execute();


    }

}
